package com.sh.data.engine.web.webapi.controller.datadev;

import com.sh.data.engine.application.model.dto.datadev.FlinkDatabaseResponse;
import com.sh.data.engine.application.model.request.datadev.*;
import com.sh.data.engine.application.service.datadev.FlinkTaskApplicationService;
import com.sh.data.engine.common.enumDefinition.ProjectStatisticTypeEnum;
import com.sh.data.engine.domain.annotation.LogExecution;
import com.sh.data.engine.domain.annotation.ProjectStatisticExecution;
import com.sh.data.engine.domain.annotation.userlog.Business;
import com.sh.data.engine.domain.annotation.userlog.EngineOperation;
import com.sh.data.engine.domain.base.model.PageResult;
import com.sh.data.engine.domain.datadev.flink.model.domain.*;
import com.sh.data.engine.domain.shims.db.model.PreviewDataDomain;
import com.sh.data.engine.web.webapi.controller.base.response.RestResponseEntity;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;

import javax.validation.Valid;
import java.util.List;
import java.util.Set;

@Validated
@RestController
@RequestMapping("/flink/sql")
@Tag(name = "实时开发SQL任务", description = "实时开发SQL任务API")
@Slf4j
public class FlinkTaskController {

    @Autowired
    private FlinkTaskApplicationService flinkTaskApplicationService;

    @PostMapping("/saveFlinkTaskNode")
    @LogExecution(operation = EngineOperation.ADD_UPDATE, business = Business.实时任务)
//    @Log(
//        business = Business.实时任务,
//        operation = Operation.ADD_UPDATE)
    @Operation(summary = "保存FlinkTask节点")
    @ProjectStatisticExecution(type = ProjectStatisticTypeEnum.DEVELOP_TASK_ONLINE)
////    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<Boolean> saveFlinkTaskNode(@Valid SavaFlinkTaskNodeRequest request) {
        var result = flinkTaskApplicationService.saveTaskNode(request);
        return RestResponseEntity.success(result);
    }

    @PostMapping("/saveSqlTaskScript")
    @Operation(summary = "保存脚本内容")
    @LogExecution(operation = EngineOperation.ADD_UPDATE, business = Business.实时任务脚本)
//    @Log(
//        business = Business.实时任务脚本,
//        operation = Operation.ADD_UPDATE)
////    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<Boolean> saveSqlTaskScript(
        @Valid @RequestBody SaveScriptRequest request) {
        var result = flinkTaskApplicationService.saveSqlTaskScript(request);
        return RestResponseEntity.success(result);
    }

    @PostMapping("/saveSource")
    @Operation(summary = "保存源表")
////    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<Boolean> saveSource(@Valid @RequestBody SaveSourceRequest request) {
        var result = flinkTaskApplicationService.saveSource(request);
        return RestResponseEntity.success(result);
    }

    @PostMapping("/saveSink")
    @Operation(summary = "保存结果表")
////    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<Boolean> saveSink(@Valid @RequestBody SaveSinkRequest request) {
        var result = flinkTaskApplicationService.saveSink(request);
        return RestResponseEntity.success(result);
    }

    @PostMapping("/saveDim")
    @Operation(summary = "保存维度表")
////    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<Boolean> saveDim(@Valid @RequestBody SaveDimRequest request) {
        var result = flinkTaskApplicationService.saveDim(request);
        return RestResponseEntity.success(result);
    }

    @PostMapping("/saveEnv")
    @Operation(summary = "保存环境参数")
////    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<Boolean> saveEnv(@Valid @RequestBody SaveEnvRequest request) {
        var result = flinkTaskApplicationService.saveEnv(request);
        return RestResponseEntity.success(result);
    }

    @PostMapping("/saveSchedule")
    @Operation(summary = "保存调度配置")
////    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<Boolean> saveSchedule(@Valid @RequestBody SaveScheduleRequest request) {
        var result = flinkTaskApplicationService.saveSchedule(request);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getTaskDetail")
    @Operation(summary = "获取任务详情")
////    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<FlinkTaskDetailDomain> getTaskDetail(
        @Parameter(description = "节点id", required = true) @RequestParam Long nodeId,
        @Parameter(description = "版本id") @RequestParam(required = false) Long versionId) {
        var result = flinkTaskApplicationService.getTaskDetail(nodeId, versionId);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getSqlTaskScrpit")
    @Operation(summary = "获取脚本内容")
////    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<String> getSqlTaskScrpit(
        @Parameter(description = "节点id", required = true) @RequestParam Long nodeId,
        @Parameter(description = "版本id") @RequestParam(required = false) Long versionId) {
        var result = flinkTaskApplicationService.getSqlTaskScript(nodeId, versionId);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getSourceDetail")
    @Operation(summary = "获取源表配置")
////    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<FlinkSourceDomain>> getSourceDetail(
        @Parameter(description = "节点id", required = true) @RequestParam Long nodeId,
        @Parameter(description = "版本id") @RequestParam(required = false) Long versionId) {
        var result = flinkTaskApplicationService.getSourceDetail(nodeId, versionId);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getSinkDetail")
    @Operation(summary = "获取结果表配置")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<FlinkSinkDomain>> getSinkDetail(
        @Parameter(description = "节点id", required = true) @RequestParam Long nodeId,
        @Parameter(description = "版本id") @RequestParam(required = false) Long versionId) {
        var result = flinkTaskApplicationService.getSinkDetail(nodeId, versionId);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getDimDetail")
    @Operation(summary = "获取维表配置")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<FlinkDimDomain>> getDimDetail(
        @Parameter(description = "节点id", required = true) @RequestParam Long nodeId,
        @Parameter(description = "版本id") @RequestParam(required = false) Long versionId) {
        var result = flinkTaskApplicationService.getDimDetail(nodeId, versionId);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getEnvDetail")
    @Operation(summary = "获取环境参数配置")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<String> getEnvDetail(
        @Parameter(description = "节点id", required = true) @RequestParam Long nodeId,
        @Parameter(description = "版本id") @RequestParam(required = false) Long versionId) {
        var result = flinkTaskApplicationService.getEnvDetail(nodeId, versionId);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getScheduleDetail")
    @Operation(summary = "获取调度配置")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<FlinkScheduleDomain> getScheduleDetail(
        @Parameter(description = "节点id", required = true) @RequestParam Long nodeId,
        @Parameter(description = "版本id") @RequestParam(required = false) Long versionId) {
        var result = flinkTaskApplicationService.getScheduleDetail(nodeId, versionId);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getSqlTaskVersion")
    @Operation(summary = "获取FlinkSqlTask历史版本")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<PageResult<FlinkSqlTaskVersionDomain>> getSqlTaskVersion(
        @Parameter(description = "节点id", required = true) @RequestParam Long id,
        @Parameter(description = "第几页") @RequestParam(required = false, defaultValue = "1") Integer pageNum,
        @Parameter(description = "每页大小") @RequestParam(required = false, defaultValue = "15")
        Integer pageSize) {
        var result = flinkTaskApplicationService.getSqlTaskVersion(id, pageNum, pageSize);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getDsType")
    @Operation(summary = "获取数据源类型")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<String>> getDsType(
        @Parameter(description = "业务类型，0-flinkSqlTask来源 1-flinkSqlTask目标 2-flinkSqlTask维表", required = true)
        @RequestParam
        Integer businessType) {
        var result = flinkTaskApplicationService.getDsType(businessType);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getDatabasesByDsType")
    @Operation(summary = "获取数据源")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<FlinkDatabaseResponse>> getDatabasesByDsType(
        @Parameter(description = "数据源类型", required = true) @RequestParam String dsType) {
        var result = flinkTaskApplicationService.getDatabasesByDsType(dsType);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getTopicAndColumns")
    @Operation(summary = "获取数据源的topic和字段信息")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<Set<String>> getTopicAndColumns(
        @Parameter(description = "数据源id", required = true) @RequestParam Long dsId) {
        var result = flinkTaskApplicationService.getTopicAndColumns(dsId);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getTables")
    @Operation(summary = "获取数据源的表列表")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<FlinkTableDomain>> getTables(
        @Parameter(description = "数据源id", required = true) @RequestParam Long dsId,
        @Parameter(description = "schema") @RequestParam(required = false) String schema,
        @Parameter(description = "1 - 获取分区表 0-获取所有表") @RequestParam(required = false) Integer isPartition) {
        var result = flinkTaskApplicationService.getTables(dsId, schema, isPartition);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getTableFields")
    @Operation(summary = "获取数据源的字段列表")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<FlinkTableFieldDomain>> getTableFields(
        @Parameter(description = "数据源id", required = true) @RequestParam Long dsId,
        @Parameter(description = "表名", required = true) @RequestParam(required = false)
        String tableName) { // required改为false是为了适配kafka数据源,因为其直接绑定到了topic
        var result = flinkTaskApplicationService.getTableFields(dsId, tableName);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getHbaseColumnFamilyNames")
    @Operation(summary = "获取hbase的column family名字")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<String>> getHbaseColumnFamilyNames(
        @Parameter(description = "数据源id", required = true) @RequestParam Long dsId,
        @Parameter(description = "表名", required = true) @RequestParam String tableName) {
        var result = flinkTaskApplicationService.getHbaseColumnFamilyNames(dsId, tableName);
        return RestResponseEntity.success(result);
    }

    @PostMapping("/rollbackVersion")
    @Operation(summary = "回滚版本")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<Boolean> rollbackVersion(
        @Parameter(description = "节点id", required = true) @RequestParam Long nodeId,
        @Parameter(description = "版本id", required = true) @RequestParam Long versionId) {
        var result = flinkTaskApplicationService.rollbackVersion(nodeId, versionId);
        return RestResponseEntity.success(result);
    }

    @PostMapping("/checkGrammar")
    @Operation(summary = "语法检查")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<FlinkCheckGrammarDomain> checkGrammar(
        @Parameter(description = "sql语句", required = true) @RequestParam String sql) {
        var result = flinkTaskApplicationService.checkGrammar(sql);
        var response = RestResponseEntity.success(result);
        if (result.isCorrect()) {
            response.setMsg("语法正确");
        } else {
            response.setMsg("语法错误");
        }
        return response;
    }

    @GetMapping("/getPartitionTableList")
    @Operation(summary = "获取所有分区&&内部表")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<HiveTableInfoDomain>> getPartitionTableList() {
        var result = flinkTaskApplicationService.getPartitionTableList();
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getPartitionInfoByTblName")
    @Operation(summary = "获取该表下所有分区信息")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<String>> getPartitionInfoByTblName(
        @Parameter(description = "Table name", required = true) @RequestParam String tableName) {
        var result = flinkTaskApplicationService.getPartitionInfoByTblName(tableName);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getSqlTaskSourceTables")
    @Operation(summary = "获取源表配置的所有映射表")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<FlinkSourceTableInfoDomain>> getSqlTaskSourceTables(
        @Parameter(description = "节点id", required = true) @RequestParam Long nodeId) {
        var result = flinkTaskApplicationService.getSqlTaskSourceTables(nodeId);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/downloadSqlTaskTemplate")
    @Operation(summary = "下载模版")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public void downloadSqlTaskTemplate(
        @Parameter(description = "节点id", required = true) @RequestParam Long nodeId,
        @Parameter(description = "表名", required = true) @RequestParam String tableName,
        HttpServletResponse response) {
        flinkTaskApplicationService.downloadSqlTaskTemplate(nodeId, tableName, response);
    }

    @PostMapping("/uploadDebugData")
    @Operation(summary = "上传测试数据")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<List<String>>> uploadDebugData(
        HttpServletRequest request, @RequestPart("file") MultipartFile file) {
        var result = flinkTaskApplicationService.uploadDebugData(file, request.getParameterMap());
        return RestResponseEntity.success(result);
    }

    @PostMapping("/collectOnlineData")
    @Operation(summary = "采集线上数据，针对kafka数据源")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<List<String>>> collectOnlineData(
        @Parameter(description = "节点id", required = true) @RequestParam Long nodeId,
        @Parameter(description = "表名", required = true) @RequestParam String tableName,
        @Parameter(description = "采集条数", required = true) @RequestParam Integer maxCount) {
        var result = flinkTaskApplicationService.collectOnlineData(nodeId, tableName, maxCount);
        return RestResponseEntity.success(result);
    }

    @PostMapping("/startDebug")
    @Operation(summary = "开始调试")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<FlinkDebugDomain> startDebug(
        @Parameter(description = "节点id", required = true) @RequestParam Long nodeId) {
        var result = flinkTaskApplicationService.startDebug(nodeId);
        return RestResponseEntity.success(result);
    }

    @PostMapping("/stopDebug")
    @Operation(summary = "停止调试")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<Boolean> stopDebug(
        @Parameter(description = "job id", required = true) @RequestParam Long jobId) {
        var result = flinkTaskApplicationService.stopDebug(jobId);
        return RestResponseEntity.success(result);
    }

    @GetMapping("/getLog")
    @Operation(summary = "获取日志")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<FlinkJobLogDomain> getLog(
        @Parameter(description = "任务id", required = true) @RequestParam Long jobId) {
        var result = flinkTaskApplicationService.getLog(jobId);
        return RestResponseEntity.success(result);
    }

    @Operation(summary = "预览数据，返回最多10行记录")
    @GetMapping("/previewData")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<PreviewDataDomain> previewData(
        @Parameter(description = "数据源id", required = true) @RequestParam Long dsId,
        @Parameter(description = "表名", required = true) @RequestParam String tableName) {
        var result = flinkTaskApplicationService.previewData(dsId, tableName);
        return RestResponseEntity.success(result);
    }

    @Operation(summary = "预览kafka数据，返回最多10行记录")
    @GetMapping("/previewKafkaData")
//    @AuthorityAnnotation(menuId = MenuEnumNew.SSKF, readWriteStatus = 1, isAdd = 1)
    public RestResponseEntity<List<String>> previewKafkaData(
        @Parameter(description = "数据源id", required = true) @RequestParam Long dsId,
        @Parameter(description = "topic", required = true) @RequestParam String topic) {
        var result = flinkTaskApplicationService.previewKafkaData(dsId, topic);
        return RestResponseEntity.success(result);
    }
}
